package com.central.face.streaming

import java.util
import java.util.{Properties, UUID}

import com.alibaba.fastjson.{JSON, JSONObject}
import com.central.utils._
import com.central.utils.es.ESMethod
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql._
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}
import org.elasticsearch.spark.rdd.EsSpark

object FaceStreaming {
  val FACE_URL: String = ConfigUtil.getProperties().getProperty("mysql.url")
  val TABLE_RESULT: String = ConfigUtil.getProperties().getProperty("mysql.face.table.result")
  val TABLE_DETAIL: String = ConfigUtil.getProperties().getProperty("mysql.face.table.detail")
  val SAVE_ES_INDEX: String = ConfigUtil.getProperties().getProperty("face.save.es.index")


  def main(args: Array[String]): Unit = {
    val configs = ConfigUtil.getProperties()
    val KAFKA_SERVERS = configs.getProperty("bootstrap.servers")
    val KAFKA_GROUPID = configs.getProperty("group.id")
    val OFFSET = configs.getProperty("auto.offset.reset")
    val FACE_TOPIC = configs.getProperty("face.topic")
    val mysql_username = configs.getProperty("mysql.username")
    val mysql_password = configs.getProperty("mysql.password")
    val elasticsearch_host = configs.getProperty("elasticsearch.host", "192.168.5.180")
    val elasticsearch_port = configs.getProperty("elasticsearch.port", "9200")

    val conf = new SparkConf().setAppName("FaceStreaming")
//      .setMaster("spark://192.168.5.180:7077")
      .setMaster("local[8]")
      .set("es.nodes", elasticsearch_host)
      .set("es.port", elasticsearch_port)
      .set("es.mapping.date.rich", "false")
      //      .set("index.refresh_interval", "60s")
      //      .set("index.translog.durability", "async")
      //      .set("index.translog.sync_interval", "60s")
      //      .set("es.batch.size.entries", "10000")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .registerKryoClasses(Array(classOf[String], classOf[SaveEsData]))


    val sc = new SparkContext(conf)
    sc.setLogLevel("ERROR")
    val ssc = new StreamingContext(sc, Seconds(120))
    val session = SparkSession.builder().config(conf).getOrCreate()


    val properties = new Properties()
    properties.put("user", mysql_username)
    properties.put("password", mysql_password)
    val prop = sc.broadcast(properties)
    val deviceParam: Broadcast[Dataset[Row]] = ssc.sparkContext.broadcast(TransData.transFaceData(session))


    val kafkaparam = Map(
      "bootstrap.servers" -> KAFKA_SERVERS,
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> KAFKA_GROUPID,
      "auto.offset.reset" -> OFFSET,
      //      "auto.offset.reset" -> "erliest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )
    val topics = Array(FACE_TOPIC)


    val stream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](topics, kafkaparam)
    )

    stream.map(x => {
      val str = x.value()

      val nObject = JSON.parseObject(str)
      val device_id=nObject.getString("device_code")
      val face_id=nObject.getString("face_id")
      val time=nObject.getString("captime")
      val address=nObject.getString("address")
      val deviceName=nObject.getString("device_name")
      val longitude=nObject.getString("longitude")
      val latitude=nObject.getString("latitude")
      val facepath=nObject.getString("face_path")
      (device_id,face_id,time,address,deviceName,longitude,latitude,facepath)
//      val strings = str.split("_")
      //      deviceId + "_" + face + "_" + time + "_" + address + "_" + deviceName + "_" + longitude + "_" + latitude
//      (strings(0), strings(1), strings(2), strings(3), strings(4), strings(5), strings(6),strings(7))
    }).filter(x => {
      (x._3.length == 13)
    })
      .foreachRDD(rdds => {
        val cacherdds = rdds.cache()
        val faceRdd = cacherdds
          .map(x => {
            (x._1, x._2 + "\\u001" + x._3 + "\\u001" + x._4 + "\\u001" + x._5 + "\\u001" + x._6 + "\\u001" + x._7+"\\u001"+x._8)
          })

        //(siteId,time)
        //        广播变量的值为(device_code,siteId)与人像数据(device_code,详情)关联得到(siteId,详情)
        val site_face = deviceParam.value.rdd
          .map(x => {
            (x.getLong(0) + "", x.getLong(1) + "")
          })
          .join(faceRdd)
          .map(x => {
            x._2
          }).cache()


        //广播变量的值为(siteId,device_code)与人像数据(siteId,详情)进行join后得到(device_code,time+详情)
        val deviceCode_time: RDD[(String, String)] = deviceParam.value.rdd.map(x => {
          (x.getLong(1) + "", x.getLong(0) + "")
        })
          .join(site_face)
          .map(x => {
            x._2
          }).cache()


        //        根据人像数据中的点位和时间信息，查询es得到侦码数据-------
        if (deviceCode_time.count() > 0) {

          //根据(device_code,time+详情)查询出es中的数据
          val esRddNew = LoadESData.loadDataFromES(deviceCode_time)
            .coalesce(8, false)
            .cache()

          println(esRddNew.count())
          //查询出来的es数据转换为(device_code,详情)，再与广播变量(device_code,siteId)进行关联，
          // 使每条详情与siteId关联，进而通过siteId与人像数据关联。
          val code_esRdd = esRddNew.map(x => {
            (x._1, x)
          })
          //得到(siteId,es数据详情)
          val site_esDetailRdd = deviceParam.value.rdd.map(x => {
            (x.getLong(0) + "", x.getLong(1) + "")
          })
            .join(code_esRdd)
            .map(x => {
              val site = x._2._1
              val detail = x._2._2
              (site, detail._1 + "\\u001" + detail._2 + "\\u001" + detail._3 + "\\u001" + detail._4 + "\\u001" + detail._5 + "\\u001" + detail._6 + "\\u001" + detail._7 + "\\u001" + detail._8+"\\u001"+detail._9)
            })
            .cache()

          //通过人像数据与侦码数据进行关联，得到关联数据

          val results: RDD[(String, (String, Option[String]))] = site_face.leftOuterJoin(site_esDetailRdd)
          val joinRdd = results
            .filter(joinResult => {
              joinResult._2._2.isDefined
            })
            .map(x => {
              val faceDetail = x._2._1
              val codeDetail = x._2._2.get

              val face_message: Array[String] = faceDetail.split("\\\\u001")
              val face = face_message(0)
              val face_captime = face_message(1)
              val face_address = face_message(2)
              val face_deviceName = face_message(3)
              val face_longitude = face_message(4)
              val face_latitude = face_message(5)
              val face_path = face_message(6)

              val code_message: Array[String] = codeDetail.split("\\\\u001")
              val deviceId = code_message(0)
              val code = code_message(1)
              val code_types = code_message(2)
              val code_capTime = code_message(3)
              val code_address = code_message(4)
              val code_deviceName = code_message(5)
              val code_longitude = code_message(6)
              val code_latitude = code_message(7)
              val es_id = code_message(8)
              (face + "_" + code, //1
                deviceId, //2
                code_types, //3
                code_capTime, //4
                code_address,
                code_deviceName,
                code_longitude,
                code_latitude,
                face_captime,
                face_address,
                face_deviceName,
                face_longitude,
                face_latitude
                ,face_path
              ,es_id
              )
            })
            .sortBy(x => {
              x._4
            })

          joinRdd.count()

          //          val esRdd = LoadESData.loadDataFromES(cacherdds.map(x => {
          //            (x._1, x._3)
          //          }))
          //            .coalesce(8, false)
          //            .cache()
          //          val faceClueRdd: RDD[(String, String)] = esRdd.map(esrdd => {
          //
          //            //deviceId, code type captime address deviceName longitude latitude
          //            (esrdd._1, esrdd._2 + "_" + esrdd._3 + "_" + esrdd._4 + "_" + esrdd._5 + "_" + esrdd._6 + "_" + esrdd._7 + "_" + esrdd._8) //(deviceId,mac/imsi,type)
          //          })
          //          val praperRdd = cacherdds.map(x => {
          //            //deviceId,face_captime_address_deviceName_longitude_latitude
          //            (x._1, x._2 + "_" + x._3 + "_" + x._4 + "_" + x._5 + "_" + x._6 + "_" + x._7)
          //          })

          //通过人像数据与侦码数据进行关联，得到关联数据
          //          val result: RDD[(String, (String, Option[String]))] = praperRdd.leftOuterJoin(faceClueRdd)
          //
          //          val joinRdd = result
          //            .filter(joinResult => {
          //              joinResult._2._2.isDefined
          //            })
          //            .map(
          //              filterResult => {
          //                //deviceId,  mac/imsi,  id
          //                val deviceId = filterResult._1
          //                val codeMessage = filterResult._2._2.get
          //                val faceIdAndTime = filterResult._2._1
          //
          //                val code_message = codeMessage.split("_")
          //                val code = code_message(0)
          //                val code_types = code_message(1)
          //                val code_capTime = code_message(2)
          //                val code_address = code_message(3)
          //                val code_deviceName = code_message(4)
          //                val code_longitude = code_message(5)
          //                val code_latitude = code_message(6)
          //
          //                val face_message = faceIdAndTime.split("_")
          //                val face = face_message(0)
          //                val face_captime = face_message(1)
          //                val face_address = face_message(2)
          //                val face_deviceName = face_message(3)
          //                val face_longitude = face_message(4)
          //                val face_latitude = face_message(5)
          //
          //                (face + "_" + code, //1
          //                  deviceId, //2
          //                  code_types, //3
          //                  code_capTime, //4
          //                  code_address,
          //                  code_deviceName,
          //                  code_longitude,
          //                  code_latitude,
          //                  face_captime,
          //                  face_address,
          //                  face_deviceName,
          //                  face_longitude,
          //                  face_latitude)
          //              }).sortBy(x => {
          //            x._4
          //          })

          //将关联数据存储到es
          val saveEsRdd = joinRdd.map(x => {
            //faceAndCode,face,code,time,place
            SaveEsData(x._1, x._2, x._3, x._4, x._5, x._6, x._7, x._8, x._9, x._10, x._11, x._12, x._13,
              x._14

//              , x._1 + x._9
            ,x._15
            )
          })
            .distinct()
            .repartition(8)
            .cache()

          import session.implicits._

          //通过redis对关联信息进行计数
          //flag, key, count, face_captime, codeType, savedata
          val accomulatedRdd: RDD[(Boolean, String, String, String, String, SaveEsData, String)] = saveEsRdd.map(x => {
            RedisUtil.accomulatorByString(x.face_code, x.face_captime, x.code_types, x, x.code_capTime)
          })
            .cache()

          val saveESRdd = accomulatedRdd
            .filter(x => {
              !x._1
            })
            .cache()


          EsSpark.saveToEs(saveESRdd.map(x => {
            val saveData: SaveEsData = x._6
            val json = new JSONObject()
            val face_code = saveData.face_code.split("_")
            json.put("faceid", face_code(0))
            json.put("tag_code", face_code(1))
            json.put("facepath",saveData.face_path)
//            json.put("deviceId", saveData.deviceId)
//            json.put("code_types", saveData.code_types)
            json.put("captime", saveData.code_capTime)
//            json.put("code_address", saveData.code_address)
//            json.put("code_deviceName", saveData.code_deviceName)
//            json.put("code_longitude", saveData.code_longitude)
//            json.put("code_latitude", saveData.code_latitude)
//            json.put("face_captime", saveData.face_captime)
//            json.put("face_address", saveData.face_address)
//
//            json.put("face_deviceName", saveData.face_deviceName)
//            json.put("face_longitude", saveData.face_longitude)
//            json.put("face_latitude", saveData.face_latitude)
            json.put("dataid", saveData.id)
            json

          }), SAVE_ES_INDEX)


          val notSaveEsRdd = accomulatedRdd.filter(x => {
            x._1
          })
          val insert = saveESRdd.filter(x => {
            x._3.equals("5")
          })
          if (insert.count() > 0) {
            insert.map(x=>{

              x._6.code_types
              val face_code = x._6.face_code.split("_")
              InsertData (face_code(0),face_code(1),x._6.code_types,x._3.toInt,x._6.face_path,50.00,System.currentTimeMillis())
            })
              .toDF().write.mode(SaveMode.Append).jdbc(FACE_URL, TABLE_RESULT, prop.value)

//=================================================旧结果表格式代码

//            val insertRdd = LoadESData.loadHistory(insert)
//            insertRdd.map(x => {
//              (x._1, x._3 + "_" + x._4)
//            })
//              .groupByKey()
//              .map(x => {
//                val face_code = x._1.split("_")
//                val count = x._2.size + ""
//                val times: Iterable[String] = x._2
//                val code_type = times.map(x => {
//                  val type_times = x.split("_")
//                  type_times(0)
//                }).toList.head
//                val time = times
//                  .map(x => {
//                    x.split("_")(1)
//                  })
//                  .toList.sortBy(x => {
//                  x
//                })
//                val firstTime = time.head
//                val lastTime = time(time.length - 1)
//                val face = face_code(0)
//                val code = face_code(1)
//                InsertData(face, code, code_type, count.toInt,
//
////                  ParseTime.DateToLong(firstTime), ParseTime.DateToLong(lastTime),
//                  50.00)
//              })
//              .toDF().write.mode(SaveMode.Append).jdbc(FACE_URL, TABLE_RESULT, prop.value)

 //=================================================




//==================================================旧详情表存储代码
//            insertRdd.map(x => {
//
//              SaveMysqlData(
//                x._1.split("_")(0),
//                x._1.split("_")(1),
//                x._2.toInt,
//                x._3,
//                ParseTime.DateToLong(x._4),
//                x._5, x._6, x._7, x._8, ParseTime.DateToLong(x._9), x._10, x._11, x._12, x._13, x._2.toInt
//              )
//            })
//              .toDF().write.mode(SaveMode.Append).jdbc(FACE_URL, TABLE_DETAIL, prop.value)
//===================================================
          }
          //达到阈值的数据进行更新操作
          if (notSaveEsRdd.count() > 0) {
            val updataParamRdd = notSaveEsRdd.filter(x => {
              !x._3.equals("5")
            }).cache()

            if (updataParamRdd.count() > 0) {
              //              val update: Array[(String, String, String, String)] =
              var update = updataParamRdd
                .map(x => {
                  //            x      (flag, key, count, face_captime, codeType, savedata, code_captime)
                  val key = x._2
                  val count = x._3
                  val time = x._4
                  val code_type = x._5
                  (key, count, time, code_type)
                  //                SaveData.updateResult(key, count, time, code_type)
                })
                .groupBy(x => {
                  x._1
                })
                .distinct()
                .map(
                  x => {

                    val head: (String, String, String, String) = x._2.toArray.sortBy(x => {
                      Integer.parseInt(x._2)
                    }).reverse.head
                    head
                  })
                .cache()

//                .map(x => {
//                  //
//
//                  SaveData.updateResultDistribute(x)
//                  //                  x
//                })
//                .count()
                                            .collect()
                                          SaveData.updateResult(update)


//=====================================================
//              updataParamRdd.map(x => {
//                x._6
//              })
//                .map(x => {
//                  SaveMysqlData(
//                    x.face_code.split("_")(0),
//                    x.face_code.split("_")(1),
//                    x.deviceId.toInt,
//                    x.code_types, ParseTime.DateToLong(x.code_capTime), x.code_address, x.face_deviceName, x.code_longitude, x.code_latitude,
//                    ParseTime.DateToLong(x.face_captime), x.face_address, x.face_deviceName, x.face_longitude, x.face_latitude, x.deviceId.toInt
//                  )
//                })
//                .toDF().write.mode(SaveMode.Append).jdbc(FACE_URL, TABLE_DETAIL, prop.value)
 // ============================================
            }
          }
        }
      })
    ssc.start()
    ssc.awaitTermination()
  }
}

case class FaceData(id: String, time: String, deviceId: String, place: String)

case class SaveEsData(face_code: String,
                      deviceId: String,
                      code_types: String,
                      code_capTime: String,
                      code_address: String,
                      code_deviceName: String,
                      code_longitude: String,
                      code_latitude: String,
                      face_captime: String,
                      face_address: String,
                      face_deviceName: String,
                      face_longitude: String,
                      face_latitude: String,
                      face_path:String,
                      id: String)

case class SaveMysqlData(faceid: String, //1
                         tagcode: String, //2
                         face_deviceid: Integer, //3
                         code_type: String, //4
                         code_captime: java.lang.Long, //5
                         code_address: String, //6
                         code_device_name: String, //7
                         code_longitude: String, //8
                         code_latitude: String, //9
                         face_captime: java.lang.Long, //10
                         face_address: String, //11
                         face_device_name: String, //12
                         face_longitude: String, //13
                         face_latitude: String, //14
                         code_deviceid: Integer)

//15

case class InsertData(faceid: String,
                      tagcode: String,
                      codetype: String,
                      counts: Integer,
                      face_path:String,
//                      first_time: java.lang.Long,
//                      last_time: java.lang.Long,
                      confidence: Double,
                      create_time:java.lang.Long

                     )